package com.zyx.flinkdemo.sql.tableapi;

import com.zyx.flinkdemo.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;
import org.apache.flink.types.Row;

/**
 * @author zyx
 * @since 2021/5/22 17:24
 * desc: 将Flink的流转转化为动态表
 */
public class StreamToTableDemo {
    public static void main(String[] args) throws Exception {
        // 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 读取端口数据创建流并转化为WaterSensor Pojo
        SingleOutputStreamOperator<WaterSensor> waterSensorDs = env
                .socketTextStream("linux201", 8888)
                .map(data -> {
                    String[] split = data.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });

        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 将流转换为动态表
        Table sensorTable = tableEnv.fromDataStream(waterSensorDs);

        // 使用TableAPI过滤出id为“ws_001”的数据
        Table selectTable = sensorTable
                .where($("id").isEqual("ws_001"))
                .select($("id"), $("ts"), $("vc"));

        /*// 过时写法
        Table selectTable = sensorTable
                .where("id = 'ws_001'")
                .select("id,ts,vc");*/

        // 将selectTable转化为流进行输出
        DataStream<Row> rowDataStream = tableEnv.toAppendStream(selectTable, Row.class);
        rowDataStream.print();

        // 执行任务
        env.execute();


    }
}
